/**
 * Copyright (C), 2020-2020, 软件公司
 * FileName: ConfirmSend3
 * Author:   cakin
 * Date:     2020/4/25
 * Description: 生产者：异步confirm模式
 */
package com.rabbitmq.productconfirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * @ClassName: ConfirmSend3
 * @Description: 生产者：异步confirm模式
 * @Date: 2020/4/25
 * @Author: cakin
 */
@Slf4j
public class ConfirmSend3 {
    /**
     * 队列名
     */
    private static final String QUEUE_NAME = "test_queue_confirm1";
    /**
     * 常量
     */
    private static final int NUM20 = 20;

    public static void main( String[] args ) throws IOException, TimeoutException, InterruptedException {
        // 获取连接
        Connection connection = ConnectionUtils.getConnection();
        // 从连接开一个通道
        Channel channel = connection.createChannel();
        // 声明一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        channel.confirmSelect(); // 开始confirm模式

        // 未确认的消息列表
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        channel.addConfirmListener(new ConfirmListener() {

            // 有效的应答
            public void handleAck( long deliveryTag, boolean multiple ) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }

            // 有问题的应答
            public void handleNack( long deliveryTag, boolean multiple ) throws IOException {
                log.info("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                // TODO 处理未确认的应答
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });


        for (int i = 0; i < NUM20; i++) {
            long nextSeqNo = channel.getNextPublishSeqNo();
            String message = "hello, confirm message " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            confirmSet.add(nextSeqNo);
        }
    }
}

